package com.shujia.flink.sink

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._

object Demo3WcSInkMysql {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    val countDS: DataStream[(String, Int)] = linesDS.flatMap(_.split(","))
      .map((_, 1))
      .keyBy(_._1)
      .sum(1)

    countDS.addSink(new RichSinkFunction[(String, Int)] {

      var con: Connection = _

      override def open(parameters: Configuration): Unit = {
        //1、加载驱动
        Class.forName("com.mysql.jdbc.Driver")
        //创建链接
        con = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata?useUnicode=true&characterEncoding=utf-8", "root", "123456")

      }

      override def close(): Unit = con.close()

      override def invoke(kv: (String, Int), context: SinkFunction.Context[_]): Unit = {
        //replace 如果主键不存在就插入，如果主键存在就替换
        val stat: PreparedStatement = con.prepareStatement("replace into word_count(word,c) values(?,?)")

        stat.setString(1, kv._1)
        stat.setInt(2, kv._2)

        stat.execute()

      }
    })

    env.execute()
  }

}
